package market

import (
	"context"
	"fmt"
	"log"
	"math"
	"sync"
	"time"

	"monitor.ilin.cc/config"
	"monitor.ilin.cc/internal/models"
	"monitor.ilin.cc/internal/services/db"
	strat "monitor.ilin.cc/internal/services/strategy"

	"github.com/amir-the-h/okex"
	"github.com/amir-the-h/okex/api"
	"github.com/amir-the-h/okex/events"
	"github.com/amir-the-h/okex/events/public"
	ws_public_requests "github.com/amir-the-h/okex/requests/ws/public"
	"monitor.ilin.cc/internal/pkg/strategy/strategies"
	"monitor.ilin.cc/internal/services/alert"
)

// 连接状态
type connState int

const (
	stateDisconnected connState = iota
	stateConnecting
	stateConnected
	stateReconnecting
)

// 重试配置
type retryConfig struct {
	initialDelay  time.Duration
	maxDelay      time.Duration
	maxAttempts   int
	backoffFactor float64
}

// Collector 结构体
type Collector struct {
	client    *api.Client
	dbService *db.Service
	config    *config.MarketConfig
	monitors  []Monitor
	okexCfg   *config.OKExConfig // 添加 OKEx 配置

	ctx       context.Context
	cancel    context.CancelFunc
	wg        sync.WaitGroup
	isRunning bool
	mu        sync.Mutex

	state      connState
	stateMu    sync.RWMutex
	retryCount int
	lastError  error

	// 消息缓冲
	tickerBuffer []*models.Ticker
	bufferMu     sync.Mutex
	bufferSize   int
	flushTicker  *time.Ticker

	// 健康检查
	healthCheck *healthChecker

	strategyMgr *strat.Manager
}

// 健康检查器
type healthChecker struct {
	lastMessageTime time.Time
	checkInterval   time.Duration
	timeout         time.Duration
	mu              sync.RWMutex
}

func newHealthChecker(checkInterval, timeout time.Duration) *healthChecker {
	return &healthChecker{
		lastMessageTime: time.Now(),
		checkInterval:   checkInterval,
		timeout:         timeout,
	}
}

func (h *healthChecker) updateLastMessageTime() {
	h.mu.Lock()
	h.lastMessageTime = time.Now()
	h.mu.Unlock()
}

func (h *healthChecker) isHealthy() bool {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return time.Since(h.lastMessageTime) <= h.timeout
}

// NewCollector 创建采集器
func NewCollector(
	marketCfg *config.MarketConfig,
	okexCfg *config.OKExConfig,
	dbService *db.Service,
	alerter alert.Service,
	monitors []Monitor,
) (*Collector, error) {
	ctx, cancel := context.WithCancel(context.Background())

	client, err := api.NewClient(ctx, okexCfg.APIKey, okexCfg.SecretKey, okexCfg.Passphrase, okex.NormalServer)
	if err != nil {
		cancel()
		return nil, err
	}

	// 使用现有的告警服务创建策略管理器
	strategyMgr := strat.NewManager(alerter)

	// 为每个订阅的交易对添加策略
	for _, sub := range marketCfg.Subscriptions {
		// 添加MA交叉策略
		maCfg := &strategies.MACrossConfig{
			FastPeriod:  5,             // 默认快速移动平均线周期为5天
			SlowPeriod:  20,            // 默认慢速移动平均线周期为20天
			MinInterval: 1 * time.Hour, // 默认最小间隔为1小时
		}
		strategyMgr.AddStrategy(strategies.NewMACrossStrategy(sub.InstID, maCfg))

		// 添加RSI策略
		rsiCfg := &strategies.RSIConfig{
			Period:      14,            // 默认周期为14天
			OverSold:    30,            // 超卖阈值 默认值30
			OverBought:  70,            // 超买阈值 默认值70
			MinInterval: 1 * time.Hour, // 默认最小间隔为1小时
		}
		strategyMgr.AddStrategy(strategies.NewRSIStrategy(sub.InstID, rsiCfg))
	}

	return &Collector{
		client:      client,
		dbService:   dbService,
		config:      marketCfg,
		monitors:    monitors,
		okexCfg:     okexCfg,
		ctx:         ctx,
		cancel:      cancel,
		bufferSize:  100,
		healthCheck: newHealthChecker(30*time.Second, time.Minute),
		strategyMgr: strategyMgr,
	}, nil
}

// 获取重试延迟
func (c *Collector) getRetryDelay() time.Duration {
	config := &retryConfig{
		initialDelay:  time.Second,
		maxDelay:      time.Minute,
		maxAttempts:   10, // 可以根据需要调整最大重试次数
		backoffFactor: 1.5,
	}

	delay := config.initialDelay * time.Duration(math.Pow(config.backoffFactor, float64(c.retryCount)))
	if delay > config.maxDelay {
		delay = config.maxDelay
	}
	return delay
}

func (c *Collector) setState(state connState) {
	c.stateMu.Lock()
	c.state = state
	c.stateMu.Unlock()
}

func (c *Collector) getState() connState {
	c.stateMu.RLock()
	defer c.stateMu.RUnlock()
	return c.state
}

func (c *Collector) Start() error {
	c.mu.Lock()
	if c.isRunning {
		c.mu.Unlock()
		return nil
	}
	c.isRunning = true
	c.mu.Unlock()

	// 启动消息处理
	c.wg.Add(2)
	go c.run()
	go c.processBuffer()

	return nil
}

func (c *Collector) Stop() {
	c.mu.Lock()
	if !c.isRunning {
		c.mu.Unlock()
		return
	}
	c.isRunning = false
	c.mu.Unlock()

	// 取消上下文会触发所有相关资源的清理
	c.cancel()
	// 等待所有 goroutine 结束
	c.wg.Wait()
}

func (c *Collector) run() {
	defer c.wg.Done()
	log.Println("启动采集器...")

	for {
		select {
		case <-c.ctx.Done():
			log.Println("采集器正在停止...")
			return
		default:
			if err := c.connect(); err != nil {
				log.Printf("WebSocket连接失败 (状态: %v): %v", c.getState(), err)
				c.setState(stateDisconnected)
				c.lastError = err

				delay := c.getRetryDelay()
				c.retryCount++

				log.Printf("计划第%d次重连，等待时间%v...", c.retryCount, delay)

				select {
				case <-c.ctx.Done():
					log.Println("重连等待期间采集器停止")
					return
				case <-time.After(delay):
					log.Printf("开始第%d次重连尝试...", c.retryCount)
					continue
				}
			}

			log.Println("连接成功，重置重试计数")
			c.retryCount = 0
		}
	}
}

func (c *Collector) connect() error {
	c.setState(stateConnecting)

	// 确保旧连接被清理
	if c.client.Ws != nil {
		//log.Println("清理旧连接...")
		// 创建新的 WebSocket 客户端
		var err error
		// 使用原始配置重新创建客户端
		c.client, err = api.NewClient(
			c.ctx,
			c.okexCfg.APIKey,
			c.okexCfg.SecretKey,
			c.okexCfg.Passphrase,
			okex.NormalServer,
		)
		if err != nil {
			log.Printf("创建新客户端失败: %v", err)
			return fmt.Errorf("创建新客户端失败: %v", err)
		}
	}

	// 建立新连接
	if err := c.client.Ws.Connect(true); err != nil {
		log.Printf("建立WebSocket连接失败: %v", err)
		return fmt.Errorf("连接WebSocket失败: %v", err)
	}
	log.Println("WebSocket连接已建立")

	// 等待连接稳定
	time.Sleep(2 * time.Second)

	// 创建新的上下文
	connCtx, cancel := context.WithCancel(c.ctx)
	defer cancel()

	// 创建新的通道，添加足够的缓冲区
	errChan := make(chan *events.Error, 100) // 添加缓冲
	subChan := make(chan *events.Subscribe, 100)
	uSubChan := make(chan *events.Unsubscribe, 100)
	logChan := make(chan *events.Login, 100)
	sucChan := make(chan *events.Success, 100)
	tkCh := make(chan *public.Tickers, 1000) // 行情通道需要更大的缓冲

	c.client.Ws.SetChannels(errChan, subChan, uSubChan, logChan, sucChan)

	// 订阅配置的产品
	log.Printf("正在订阅%d个产品...", len(c.config.Subscriptions))

	// 添加订阅重试机制
	maxRetries := 3
	subscribeSuccess := false
	for retry := 0; retry < maxRetries; retry++ {
		allSubscribed := true
		for _, sub := range c.config.Subscriptions {
			//log.Printf("正在订阅%s (第%d次尝试)...", sub.InstID, retry+1)
			if err := c.client.Ws.Public.Tickers(ws_public_requests.Tickers{InstID: sub.InstID}, tkCh); err != nil {
				//log.Printf("订阅%s失败: %v", sub.InstID, err)
				allSubscribed = false
				break
			}
			//log.Printf("成功订阅%s", sub.InstID)
		}
		if allSubscribed {
			subscribeSuccess = true
			break
		}
		if retry < maxRetries-1 {
			log.Printf("部分订阅失败，等待后重试...")
			time.Sleep(2 * time.Second)
		}
	}

	if !subscribeSuccess {
		log.Println("订阅失败，达到最大重试次数")
		return fmt.Errorf("订阅失败")
	}

	c.setState(stateConnected)
	//log.Println("连接已建立且产品订阅成功")

	// 启动健康检查
	healthTicker := time.NewTicker(c.healthCheck.checkInterval)
	defer healthTicker.Stop()

	// 重置健康检查状态
	c.healthCheck.updateLastMessageTime()

	// 添加连接监控
	lastMessageTime := time.Now()
	connectionTimeout := 30 * time.Second
	go func() {
		ticker := time.NewTicker(5 * time.Second)
		defer ticker.Stop()

		for {
			select {
			case <-connCtx.Done():
				log.Println("连接监控已停止")
				return
			case <-ticker.C:
				if time.Since(lastMessageTime) > connectionTimeout {
					log.Printf("连接监控：超过%v未收到任何消息，可能已断开", connectionTimeout)
					cancel()
					return
				}
			}
		}
	}()

	// 消息处理循环
	for {
		select {
		case <-connCtx.Done():
			log.Println("连接上下文结束，正在关闭连接...")
			// 尝试清空通道
			c.drainChannels(errChan, subChan, uSubChan, logChan, sucChan, tkCh)
			return nil
		case <-healthTicker.C:
			if !c.healthCheck.isHealthy() {
				log.Printf("连接不健康：已有%v未收到消息", c.healthCheck.timeout)
				return fmt.Errorf("连接不健康：已有%v未收到消息", c.healthCheck.timeout)
			}
			//log.Println("连接健康检查通过")
		case <-logChan:
			lastMessageTime = time.Now()
			c.healthCheck.updateLastMessageTime()
			log.Print("[已授权] WebSocket连接已授权")
		case success := <-sucChan:
			lastMessageTime = time.Now()
			c.healthCheck.updateLastMessageTime()
			log.Printf("[成功] 操作成功: %+v", success)
		case sub := <-subChan:
			lastMessageTime = time.Now()
			c.healthCheck.updateLastMessageTime()
			channel, _ := sub.Arg.Get("instId")
			log.Printf("[已订阅] 成功订阅频道: %s", channel)
		case uSub := <-uSubChan:
			lastMessageTime = time.Now()
			c.healthCheck.updateLastMessageTime()
			channel, _ := uSub.Arg.Get("channel")
			log.Printf("[取消订阅] 频道已取消订阅: %s", channel)
		case err := <-errChan:
			lastMessageTime = time.Now()
			c.healthCheck.updateLastMessageTime()
			log.Printf("[错误] 收到WebSocket错误: %+v", err)
			if c.shouldReconnect(err) {
				log.Printf("检测到严重错误，开始重连: %v", err)
				return fmt.Errorf("严重错误: %v", err)
			}
		case tickers := <-tkCh:
			lastMessageTime = time.Now()
			c.healthCheck.updateLastMessageTime()
			if len(tickers.Tickers) > 0 {
				//log.Printf("收到%s的行情数据，最新价格: %v", tickers.Tickers[0].InstID, tickers.Tickers[0].Last)
				c.bufferTickers(tickers)
			} else {
				//log.Println("收到空的行情数据")
			}
		}
	}
}

// 判断是否需要重连
func (c *Collector) shouldReconnect(err *events.Error) bool {
	// 根据错误类型判断是否需要重连
	switch err.Code {
	case 50001, 50002, 50003: // 示例错误码
		return true
	default:
		return err.Code >= 50000
	}
}

// 缓冲处理
func (c *Collector) bufferTickers(tickers *public.Tickers) {
	if tickers == nil || len(tickers.Tickers) == 0 {
		return
	}

	c.bufferMu.Lock()
	defer c.bufferMu.Unlock()

	for _, p := range tickers.Tickers {
		// 添加数据验证
		if p.InstID == "" || p.Last <= 0 || time.Time(p.TS).IsZero() {
			log.Printf("跳过无效数据: instId=%s, last=%.2f, ts=%v",
				p.InstID, p.Last, time.Time(p.TS))
			continue
		}

		ticker := &models.Ticker{
			InstType:  string(p.InstType),
			InstID:    p.InstID,
			Last:      p.Last,
			LastSz:    p.LastSz,
			AskPx:     p.AskPx,
			AskSz:     p.AskSz,
			BidPx:     p.BidPx,
			BidSz:     p.BidSz,
			Open24h:   p.Open24h,
			High24h:   p.High24h,
			Low24h:    p.Low24h,
			VolCcy24h: p.VolCcy24h,
			Vol24h:    p.Vol24h,
			SodUtc0:   p.SodUtc0,
			SodUtc8:   p.SodUtc8,
			CreatedAt: time.Time(p.TS),
		}
		c.tickerBuffer = append(c.tickerBuffer, ticker)
		//log.Printf("已缓存%s的行情数据: 价格=%v, 时间=%v", ticker.InstID, ticker.Last, ticker.CreatedAt)

		// 调用策略管理器时传入交易对信息
		c.strategyMgr.OnTick(p.InstID, float64(p.Last), time.Time(p.TS))
	}

	if len(c.tickerBuffer) >= c.bufferSize {
		//log.Printf("缓冲区已满(%d条数据)，触发刷新", len(c.tickerBuffer))
		c.flushBuffer()
	}
}

// 处理缓冲区数据
func (c *Collector) processBuffer() {
	defer c.wg.Done()

	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-c.ctx.Done():
			return
		case <-ticker.C:
			c.flushBuffer()
		}
	}
}

// 刷新缓冲区
func (c *Collector) flushBuffer() {
	c.bufferMu.Lock()
	if len(c.tickerBuffer) == 0 {
		c.bufferMu.Unlock()
		return
	}

	// 限制每次处理的数量
	maxBatchSize := 100
	var tickersToProcess []*models.Ticker
	if len(c.tickerBuffer) > maxBatchSize {
		tickersToProcess = c.tickerBuffer[:maxBatchSize]
		c.tickerBuffer = c.tickerBuffer[maxBatchSize:]
	} else {
		tickersToProcess = c.tickerBuffer
		c.tickerBuffer = make([]*models.Ticker, 0, c.bufferSize)
	}
	c.bufferMu.Unlock()

	//log.Printf("开始处理%d条行情数据", len(tickersToProcess))

	for _, ticker := range tickersToProcess {
		// 添加重试机制
		retries := 3
		var saved bool
		for i := 0; i < retries; i++ {
			if err := c.dbService.SaveTicker(ticker); err != nil {
				log.Printf("保存行情数据失败 (第%d次尝试): %v", i+1, err)
				if i < retries-1 {
					time.Sleep(time.Second * time.Duration(i+1))
					continue
				}
			} else {
				saved = true
				break
			}
		}

		if !saved {
			log.Printf("保存行情数据失败，放回缓冲区: %s", ticker.InstID)
			c.bufferMu.Lock()
			c.tickerBuffer = append(c.tickerBuffer, ticker)
			c.bufferMu.Unlock()
			// 如果保存失败，暂停一段时间再继续
			time.Sleep(5 * time.Second)
			continue
		}

		// 执行监控
		for _, monitor := range c.monitors {
			if err := monitor.OnTick(float64(ticker.Last), ticker.CreatedAt); err != nil {
				log.Printf("监控处理失败: %v", err)
			}
		}
	}

	//log.Printf("完成处理%d条行情数据", len(tickersToProcess))
}

// 添加新的方法来清理通道
func (c *Collector) drainChannels(
	errChan chan *events.Error,
	subChan chan *events.Subscribe,
	uSubChan chan *events.Unsubscribe,
	logChan chan *events.Login,
	sucChan chan *events.Success,
	tkCh chan *public.Tickers,
) {
	// 使用非阻塞方式清空通道
	for {
		select {
		case <-errChan:
		case <-subChan:
		case <-uSubChan:
		case <-logChan:
		case <-sucChan:
		case <-tkCh:
		default:
			return
		}
	}
}

// 连接状态的字符串表示
func (s connState) String() string {
	switch s {
	case stateDisconnected:
		return "已断开"
	case stateConnecting:
		return "连接中"
	case stateConnected:
		return "已连接"
	case stateReconnecting:
		return "重连中"
	default:
		return "未知状态"
	}
}
